package com.feidee.fd.sml.algorithm.component.functions

import com.alibaba.fastjson.{JSONArray, JSONObject}
import com.feidee.fd.sml.algorithm.component.{AbstractComponent, BasicParam}
import com.feidee.fdspark.transformer.{ColSelector, MetaStorage, ModelType}
import com.google.gson.Gson
import org.apache.spark.ml.{Pipeline, PipelineModel, PipelineStage}
import org.apache.spark.sql.DataFrame

import scala.collection.mutable.ArrayBuffer

/**
  * @author xiongjun
  * @date 2019/9/9 10:15
  * @description
  * @reviewer
  */
case class MultiFieldWraperParam(override val input_pt: String,
                                 override val output_pt: String,
                                 override val hive_table: String,
                                 override val flow_time: String,
                                 inputCols:Array[String],
                                 outputCols:Array[String],
                                 modelPath:String,
                                 preserveCols: String,
                                 className: String,
                                 jsonParam:String
                                ) extends BasicParam{
  def this() = this(null,null,null,null,null, null, null, null, null, null)

  override def verify(): Unit = {
    super.verify()
    require(tool.isNotNull(input_pt), "param input_pt can't be null")
    require(inputCols.length>0 && outputCols.length>0,"param inputCols and outputCols length must >0")
    require(inputCols.length==outputCols.length,"inputCols and outputCols can't match")
    require(tool.isNotNull(className),"param className can't be null")
    require(tool.isNotNull(jsonParam)&&jsonParam.startsWith("[")&&jsonParam.endsWith("]"),"param jsonParam can't be null and must valid json array")
    require(!tool.isNotNull(preserveCols) || preserveCols.split(",").intersect(outputCols).length == 0,
      "保留字段 preserveCols 不能与输出字段 outputCol 出现重合")
  }

  override def toMap: Map[String, Any] = {
    var map = super.toMap
    map += ("inputCols" -> inputCols)
    map += ("outputCols" -> outputCols)
    map += ("modelPath" -> modelPath)
    map += ("preserveCols" -> preserveCols)
    map += ("className" -> className)
    map += ("jsonParam"->jsonParam)
    map
  }
}

class MultiFieldWraper extends AbstractComponent[MultiFieldWraperParam] {

  def train(param: MultiFieldWraperParam, data: DataFrame): PipelineModel = {
    // 把 preserveCols 取出来
    val preserveCols = if (tool.isNotNull(param.preserveCols)) {
      param.preserveCols.split(",")
    } else {
      // 不填 preserveCols 时，赋上一个空数组，不然上面会返回一个长度为 1 内容空字符串的数组！
      new Array[String](0)
    }
    // 结果列
    val cols = preserveCols ++ param.outputCols
    val classnames = param.className.split("\\.")
    val modelType = classnames(classnames.length-2) match {
      case "feature"=>ModelType.Feature
      case "preprocess"=>ModelType.Preprocessing
      case _=>
        throw new Exception(s"current only support feature or preprocess,can't support ${classnames(classnames.length-2)}")
    }
    // 存储训练参数元信息
    val meta = new MetaStorage()
      .setModelType(modelType)
      .setParameters(param.toMap)
      .setFields(data.schema.fieldNames)

    // 最终结果只保留结果列
    val colSelector = new ColSelector()
      .setCols(cols)

    // 执行流程：存储元信息 》 特征组件处理过程 》 保留结果列
    val pipeline = new Pipeline().setStages(Array(meta) ++ setUp(param, data) ++ Array(colSelector))
    val model = pipeline.fit(data)

    model
  }

  def setUp(param: MultiFieldWraperParam,data: DataFrame):Array[PipelineStage] = {
//    logInfo(s"jsonParam: ${param.jsonParam}")
    val jsonArray = new Gson().fromJson(param.jsonParam,classOf[JSONArray])
    val clazz = Class.forName(param.className)
    val paramClazz = Class.forName(param.className+"Param")
    val paramFields = paramClazz.getDeclaredFields.filter(_.getName!="tool").map(_.getName)

//    logInfo(s"fields length is ${paramFields.length},value:${paramFields.mkString(",")}")
    val paramConstructor = paramClazz.getDeclaredConstructors().filter(_.getParameterCount>0).head

    val parameterTypes = paramConstructor.getParameterTypes.map(p=>p.getCanonicalName)
//    logInfo(s"parameters length is ${parameterTypes.length}")
    val fieldAndTypes = paramFields.zip(parameterTypes)

    val params = ArrayBuffer[Object]()
    for (i<- 0 until param.inputCols.length){
      val jsonObject = if (i<jsonArray.size()){
        jsonArray.getJSONObject(i)
      }else{
        new JSONObject()
      }
//      logInfo(s"$i-th jsonObject: ${jsonObject.toJSONString}")
      val orParam = Array[Object](param.input_pt,param.output_pt,param.hive_table,param.flow_time,param.inputCols(i),param.outputCols(i),param.preserveCols,param.modelPath)
      val missParams = fieldAndTypes.slice(8,fieldAndTypes.length).filter(kv=> !jsonObject.containsKey(kv._1)).map(_._1)
      require(missParams.isEmpty,s"miss params:${missParams.mkString(",")}")
      val perArray = fieldAndTypes.filter(kv=> jsonObject.containsKey(kv._1)).map(kv=>{
        val perParam = kv._2 match {
          case ParamType.STRING=>
            jsonObject.getString(kv._1)
          case ParamType.STRING_ARR=>
            jsonObject.getJSONArray(kv._1).toArray().map(_.toString)
          case ParamType.INT=>
            jsonObject.getInteger(kv._1)
          case ParamType.INT_ARR =>
            jsonObject.getJSONArray(kv._1).toArray().map(_.toString.toDouble.toInt)
          case ParamType.DOUBLE =>
            jsonObject.getDouble(kv._1)
          case ParamType.DOUBLE_ARR =>
            jsonObject.getJSONArray(kv._1).toArray().map(_.toString.toDouble)
          case ParamType.LONG =>
            jsonObject.getLong(kv._1)
          case ParamType.LONG_ARR =>
            jsonObject.getJSONArray(kv._1).toArray().map(_.toString.toLong)
          case ParamType.BOOLEAN =>
            jsonObject.getBoolean(kv._1)
          case ParamType.BOOLEAN_ARR =>
            jsonObject.getJSONArray(kv._1).toArray().map(_.toString.toBoolean)
          case _ =>
            throw new Exception("param type error")
        }
        perParam.asInstanceOf[Object]
      })
      val paramArray = orParam++perArray
      logInfo(s"paramArray lenght: ${paramArray.length}, paramArray: ${paramArray.mkString(" ")}")
      params+=paramConstructor.newInstance(paramArray:_*).asInstanceOf[Object]
    }
    val setUpMethod = clazz.getDeclaredMethod("setUp",paramClazz,classOf[DataFrame])
    val pipelineStages = params.flatMap(p=>{
      setUpMethod.invoke(clazz.newInstance(),p,data).asInstanceOf[Array[PipelineStage]]
    }).toArray
    pipelineStages
  }

  /**
    * 保存模型，除了要保存一份到参数指定路径，还需要在该路径下保存一份以时间作为后缀的模型作为备份及供线上使用
    * @param model  要保存的模型
    * @param param  参数
    */
  def outputModel(model: PipelineModel, param: MultiFieldWraperParam): Unit = {
    model.write.overwrite().save(param.modelPath)
    if (tool.isNotNull(param.flow_time)) {
      model.write.overwrite().save(s"${param.modelPath}_${param.flow_time}")
    } else {
      logWarning("未发现运行时间参数，不做模型备份处理")
    }
  }

  override def apply(paramStr: String): Unit = {
    logInfo("parsing parameter")
    val param = parseParam(paramStr)
    logInfo("validating parameter")
    param.verify()
    logInfo(s"loading input data FROM ${param.input_pt}")
    val inputData = loadData(param)
    logInfo("training feature model")
    val model = train(param, inputData)
    val result = model.transform(inputData)
    // 保存
    if (tool.isNotNull(param.output_pt)) {
      logInfo(s"saving feature result TO ${param.output_pt}")
      outputData(result, param)
    }
    if (tool.isNotNull(param.modelPath)) {
      logInfo(s"saving feature model TO ${param.modelPath}")
      outputModel(model, param)
    }
    if (tool.isNotNull(param.hive_table)) {
      logInfo(s"saving to hive table ${param.hive_table}")
      outputTable(result, param)
    }
  }
}

object MultiFieldWraper {

  def apply(paramStr: String): Unit = {
    new MultiFieldWraper()(paramStr)
  }

  def main(args: Array[String]): Unit = {
    MultiFieldWraper(args(0))
  }
}

object ParamType{
  final val INT = "int"
  final val INT_ARR = "int[]"
  final val DOUBLE = "double"
  final val DOUBLE_ARR = "double[]"
  final val STRING = "java.lang.String"
  final val STRING_ARR = "java.lang.String[]"
  final val BOOLEAN = "boolean"
  final val BOOLEAN_ARR = "boolean[]"
  final val LONG = "long"
  final val LONG_ARR = "long[]"
}